package com.g.ocache.impl;

import java.lang.Thread.UncaughtExceptionHandler;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.g.ocache.event.CacheEvent;
import com.g.ocache.event.CacheEventListener;
import com.g.ocache.event.CacheEventPublisher;
import com.g.ocache.utils.XThreadFactory;


/**
 * 
 * @author Jingqi Xu
 */
public final class AsyncCacheEventPublisher<K, V> implements CacheEventPublisher<K, V> {
	//
	private static final Logger LOGGER = LoggerFactory.getLogger(AsyncCacheEventPublisher.class);
	
	//
	private static final int DEFAULT_THREAD_COUNT = 1;
	
	//
	private final ExecutorService executor;
	private final AtomicBoolean started = new AtomicBoolean(false);
	private final CopyOnWriteArrayList<CacheEventListener<K, V>> listeners;
	
	/**
	 * 
	 */
	public AsyncCacheEventPublisher() {
		this(DEFAULT_THREAD_COUNT, null);
	}
	
	public AsyncCacheEventPublisher(int tc, ThreadFactory tf) {
		//
		if(tf == null) {
			XThreadFactory xtf = new XThreadFactory(getClass().getName(), true);
			xtf.setUncaughtExceptionHandler(new UncaughtExceptionHandler() {
				public void uncaughtException(Thread t, Throwable e) {
					LOGGER.error("unhandled exception in thread: " + t.getName(), e);
				}
			});
			tf = xtf;
		}
		this.executor = Executors.newFixedThreadPool(tc, tf);
		this.listeners = new CopyOnWriteArrayList<CacheEventListener<K, V>>();
	}
	
	public AsyncCacheEventPublisher(ExecutorService executor) {
		this.executor = executor;
		this.listeners = new CopyOnWriteArrayList<CacheEventListener<K, V>>();
	}
	
	/**
	 * 
	 */
	public boolean isRunning() {
		return started.get();
	}
	
	public void start() {
		// Precondition checking
		if(!started.compareAndSet(false, true)) {
			throw new IllegalStateException("the publisher has already been started");
		}
		
		// NOP
	}
	
	public void stop() {
		// Precondition checking
		if(!started.compareAndSet(true, false)) {
			return;
		}
		
		//
		executor.shutdownNow();
	}

	/**
	 * 
	 */
	public void publish(CacheEvent<K, V> event) {
		executor.execute(new Task(event));
	}
	
	public final boolean addCacheEventListener(CacheEventListener<K, V> listener) {
		if(listener == null || listeners.contains(listener)) {
			return false;
		} else {
			return listeners.add(listener);
		}
	}
	
	public final boolean removeCacheEventListener(CacheEventListener<K, V> listener) {
		if(listener == null) {
			return false;
		} else {
			return listeners.remove(listener);
		}
	}

	/**
	 * 
	 */
	private class Task implements Runnable {
		//
		private final CacheEvent<K, V> event;

		/**
		 * 
		 */
		public Task(CacheEvent<K, V> event) {
			this.event = event;
		}

		/**
		 * 
		 */
		public void run() {
			//
			if(listeners == null) {
				return;
			}
			
			//
			for(CacheEventListener<K, V> listener: listeners) {
				try {
					listener.onEvent(event);
				} catch(Exception e) {
					LOGGER.error("unhandled exception in cache event publisher", e);
				}
			}
		}
	}
}
